#pragma once
/*
    发布订阅功能:5个操作, 2个维护管理map, 2个内部结构体类型(订阅者, 主题)
    - 主题创建/ 删除/ 订阅主题/ 取消订阅/ 主题消息发布
    - 维护主题和订阅者的关联关系, 当有主题消息的时候, 将消息发布给订阅者 map<topic, subscriber>
    - 维护连接和订阅者的关联关系,[连接过的客户端保存其信息]
    - 当订阅者取消订阅之后, 后续不再进行消息通知 topic->removeSubscriber
    - 当订阅者下线之后, 后续不再进行通知 并且删除保存的连接信息 topic->removeSubscriber && _subscribers->removeConn
*/
#include "../common/message.hpp"
#include "../common/net.hpp"
#include <unordered_set>

namespace tmh_rpc
{
    namespace server
    {
        class TopicManager
        {
        public:
            using ptr = std::shared_ptr<TopicManager>;
            TopicManager() {}

            // 这是设置给dispatcher模块的回调函数: 根据传入的req的类型不同, (主题创建/删除/订阅..) 操作也不同
            void onTopicRequest(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                TopicOptype topic_optype = msg->opType();
                // 根据类型选择操作
                bool ret = true;
                switch (topic_optype)
                {
                    // 创建主题
                case TopicOptype::TOPIC_CREATE:
                    topicCreate(conn, msg);
                    break;

                // 删除主题
                case TopicOptype::TOPIC_REMOVE:
                    topicRemove(conn, msg);
                    break;

                // 订阅主题
                case TopicOptype::TOPIC_SUBSCRIBE:
                    ret = topicSubscribe(conn, msg);
                    break;

                // 取消订阅
                case TopicOptype::TOPIC_CANCEL:
                    topicCancel(conn, msg);
                    break;

                // 将主题消息发布给订阅者
                case TopicOptype::TOPIC_PUBLISH:
                    ret = topicPublish(conn, msg);
                    break;
                default:
                    return errorResponse(conn, msg, RCode::RCODE_INVALID_OPTYPE);
                }

                if (!ret)
                    return errorResponse(conn, msg, RCode::RCODE_NOT_FOUND_TOPIC);
                return topicResponse(conn, msg);
            }

            // 当一个订阅者在断开连接时候的回调函数 删除其关联的数据
            void onShutdown(const BaseConnection::ptr &conn)
            {
                // 消息发布者断开连接, 无需操作; 订阅者需要删除管理信息
                Subscriber::ptr subscriber;
                vector<Topic::ptr> topics;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 1. 判断当前断开的连接是否是订阅者的, [也就是处理业务的服务器]; 如果是发布者[用户], 则不需要处理

                    auto it = _subscribers.find(conn);
                    if (it == _subscribers.end())
                        return;
                    subscriber = it->second;
                    // 2. 将该订阅者从其订阅的主题中删除(在加锁范围内先获取其订阅的主题对象)
                    for (auto &topic_name : subscriber->topics)
                    {
                        auto topic_it = _topics.find(topic_name);
                        if (topic_it == _topics.end())
                            continue; // 该主题不存在
                        topics.push_back(topic_it->second);
                    }
                    // 3. 将连接映射管理信息中的该订阅者删除
                    _subscribers.erase(it);
                }
                // 从主题中删除该订阅者
                for (auto &topic : topics)
                {
                    topic->removeSubscriber(subscriber);
                }
                return;
            }

        private:
            // -------------------------------响应---------------------------
            //  当主题请求类型错误 或者找不到主题名称的时候返回错误响应信息
            void errorResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg, RCode rcode)
            {
                // 1.组织响应
                auto topic_rsp = MessageFactory::create<TopicResponse>();
                // 为了保证可靠性:响应id和请求id是一一对应的
                topic_rsp->setId(msg->rid());
                topic_rsp->setMType(MType::RSP_TOPIC);
                topic_rsp->setRCode(rcode);

                // 2.发送响应
                return conn->send(topic_rsp);
            }
            // 当主题请求成功被处理之后, 返回正确的响应信息
            void topicResponse(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1.组织响应
                auto topic_rsp = MessageFactory::create<TopicResponse>();
                // 为了保证可靠性:响应id和请求id是一一对应的
                topic_rsp->setId(msg->rid());
                topic_rsp->setMType(MType::RSP_TOPIC);
                topic_rsp->setRCode(RCode::RCODE_OK);

                // 2.发送响应
                return conn->send(topic_rsp);
            }
            // -------------------- 主题创建/ 删除/ 订阅主题/ 取消订阅/ 发布消息 -------------
            //  创建新主题
            void topicCreate(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                std::unique_lock<std::mutex> lock(_mutex);

                // 构造一个主题对象, 添加映射关系的管理
                string topic_name = msg->topicKey();
                auto topic = std::make_shared<Topic>(topic_name);
                _topics.insert({topic_name, topic});
            }
            // 删除主题
            void topicRemove(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 将该主题从_topics管理信息中删除; 2. 将订阅了该主题的订阅者取消订阅;
                string topic_name = msg->topicKey();
                std::unordered_set<Subscriber::ptr> subscribers;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    //  找到该主题是否存在
                    auto it = _topics.find(topic_name);
                    if (it == _topics.end())
                    {
                        // 不存在, 不需要进行操作
                        return;
                    }
                    // 在删除主题之前, 先获取到受影响的订阅者的信息
                    subscribers = it->second->subscribers;
                    _topics.erase(topic_name); // 删除当前的主题映射关系
                }
                for (auto &subscriber : subscribers)
                {
                    subscriber->removeTopic(topic_name);
                }
            }
            // 订阅主题
            bool topicSubscribe(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 查找该主题是否存在
                // 2. 找出主题的订阅者们是否有当前连接所对应的对象, 没有就构造一个订阅者
                Topic::ptr topic;
                Subscriber::ptr subscriber;
                string topic_name = msg->topicKey();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it == _topics.end())
                    {
                        // 不存在, 返回错误信息
                        return false;
                    }
                    topic = topic_it->second;
                    auto sub_it = _subscribers.find(conn);
                    if (sub_it != _subscribers.end()) // 说明当前客户端是一个之前发布过请求的客户端
                    {
                        subscriber = sub_it->second;
                    }
                    else
                    {
                        subscriber = std::make_shared<Subscriber>(conn); // 构造一个新的订阅者对象, 并且加入订阅者管理模块中_subscriber
                        _subscribers.insert({conn, subscriber});
                    }
                }
                // 在主题对象中, 新增一个订阅者;
                // 在订阅者对象中, 新增一个订阅的主题;
                topic->appendSubscriber(subscriber);
                subscriber->appendTopic(topic_name);
                return true;
            }
            // 取消订阅
            void topicCancel(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 1. 先找出主题对象和订阅者对象
                Topic::ptr topic;
                string topic_name = msg->topicKey();
                Subscriber::ptr subscriber;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it != _topics.end())
                        topic = topic_it->second;

                    auto sub_it = _subscribers.find(conn);
                    if (sub_it != _subscribers.end())
                    {
                        subscriber = sub_it->second;
                    }
                }
                if (subscriber)
                    subscriber->removeTopic(topic_name); // 取消订阅
                if (topic && subscriber)
                    topic->removeSubscriber(subscriber); // 删除订阅者
            }

            // 将主题信息发布给订阅者
            bool topicPublish(const BaseConnection::ptr &conn, const TopicRequest::ptr &msg)
            {
                // 找到主题的订阅者们
                Topic::ptr topic;
                string topic_name = msg->topicKey();
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto topic_it = _topics.find(topic_name);
                    if (topic_it == _topics.end())
                        return false;
                    topic = topic_it->second;
                }
                // 将消息发布给订阅者
                topic->pushMessage(msg);

                return true;
            }

        private:
            // 订阅者: 维护其内部所订阅的主题和对应连接
            struct Subscriber
            {
                using ptr = std::shared_ptr<Subscriber>;
                std::mutex _mutex;
                BaseConnection::ptr conn;
                std::unordered_set<std::string> topics; // 当前订阅者所订阅主题的主题名称

                // [一个订阅者] 一定唯一对应 [一个连接conn]
                Subscriber(const BaseConnection::ptr &c) : conn(c) {}

                // 订阅主题的时候调用
                void appendTopic(const string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.insert(topic_name);
                }

                // 主题被删除 或者 取消订阅的时候 调用
                void removeTopic(const string &topic_name)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    topics.erase(topic_name);
                }
            };
            struct Topic
            {
                using ptr = std::shared_ptr<Topic>;
                std::mutex _mutex;
                string topic_name;                               // 当前主题的名称
                std::unordered_set<Subscriber::ptr> subscribers; // 当前主题的订阅者们

                // 主题名称必须在构造的时候初始化好
                Topic(const string &name) : topic_name(name) {}

                // 新增订阅者的时候调用
                void appendSubscriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.insert(subscriber); // 往管理信息中添加新订阅者
                }

                // 取消订阅 或者订阅者连接断开的时候 调用
                void removeSubscriber(const Subscriber::ptr &subscriber)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    subscribers.erase(subscriber); // 删除订阅者的信息
                }

                // 收到消息发布请求 向订阅者发布消息
                void pushMessage(const BaseMessage::ptr &msg)
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    for (auto &subscriber : subscribers)
                    {
                        // 连接和订阅者一一对应, 通过conn向订阅者发送消息
                        subscriber->conn->send(msg);
                    }
                }
            };

        private:
            std::mutex _mutex;
            std::unordered_map<string, Topic::ptr> _topics;
            std::unordered_map<BaseConnection::ptr, Subscriber::ptr> _subscribers;
        };
    }
}